package com.snopy.mq.kafka.serial;

import com.snopy.pojo.Banana;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.DoubleDeserializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

/**
 * @author snopy
 * @version 1.0.0
 * @classname BananaDeserializer
 * @description
 * @email 77912204@qq.com
 * @date 2022/06/09 22:30
 */
public class BananaDeserializer implements Deserializer<Banana> {
    /**
     * Deserialize a record value from a byte array into a value or object.
     *
     * @param topic topic associated with the data
     * @param data  serialized bytes; may be null; implementations are recommended to handle null by returning a value or null rather than throwing an exception.
     * @return deserialized typed data; may be null
     */
    @Override
    public Banana deserialize(String topic, byte[] data) {
        if (null == data){
            return null;
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int colorLen = buffer.getInt();
        byte[] colorBytes = new byte[colorLen];
        buffer.get(colorBytes);
        String color = new String(colorBytes, StandardCharsets.UTF_8);

        int nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);
        String name = new String(nameBytes, StandardCharsets.UTF_8);

        int weightLen = buffer.getInt();
        byte[] weightBytes = new byte[weightLen];
        buffer.get(weightBytes);
        Double weight = new DoubleDeserializer().deserialize(null,weightBytes);

        int descLen = buffer.getInt();
        byte[] descBytes = new byte[descLen];
        buffer.get(descBytes);
        String desc = new String(descBytes, StandardCharsets.UTF_8);
        return new Banana(color,name,weight,desc);
    }
}
